package com.ydlclass.core;

import com.ydlclass.NettyBootStrapInitializer;
import com.ydlclass.YrpcBootstrap;
import com.ydlclass.compress.Factory.CompressFactory;
import com.ydlclass.enumeration.RequestType;
import com.ydlclass.serialize.Factory.SerializeFactory;
import com.ydlclass.transport.message.YrpcRequest;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 心跳检测的目的是什么? 探活机制感知哪些服务的连接状态是正常的，哪些是不正常的
 * 心跳检测工具
 * @Author duWenJian
 * @Date 2023/11/6 16:48
 * @Version 1.0
 */
@Slf4j
public class HeartbeatDetector {
    /**
     * 心跳检测
     */
    public static void detectHeartbeat(String serviceName){
        // 1、从注册中心拉取任务列表并建立连接
        List<InetSocketAddress> addresses = YrpcBootstrap.getInstance().getRegistry().lookup(serviceName);

        // 2、将连接进行缓存
        for (InetSocketAddress address : addresses) {
            try {
                if (!YrpcBootstrap.CHANNEL_CACHE.containsKey(address)){
                    Channel channel = NettyBootStrapInitializer.getBootstrap().connect(address).sync().channel();
                    YrpcBootstrap.CHANNEL_CACHE.put(address,channel);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

        }

        // 3、任务,定期发送消息
        Thread thread = new Thread(()->{
            new Timer().scheduleAtFixedRate(new MyTimerTask(),0,2000);
        },"yrpc_HeartbeatDetector_thread");

        thread.setDaemon(true);
        thread.start();


    }
    // 定时任务
    private static class MyTimerTask extends TimerTask {
        @Override
        public void run() {
            // 将响应时长的map进行清空
            YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.clear();

            // 1、遍历缓存的连接channel
            for (Channel channel : YrpcBootstrap.CHANNEL_CACHE.values()) {
                // 2、发送心跳
                int tryStamp = 3;
                while (tryStamp > 0) {
                long startTime = System.currentTimeMillis();
                // 构建一个心跳请求
                YrpcRequest yrpcRequest = YrpcRequest.builder()
                        .requestId(YrpcBootstrap.ID_GENERATOR.getId())
                        .compressType(CompressFactory.getCompressor(YrpcBootstrap.COMPRESSOR_TYPE).getCode())
                        .requestType(RequestType.HEART_BEAT.getId())
                        .serializeType(SerializeFactory.getSerializer(YrpcBootstrap.SERIALIZE_TYPE).getCode())
                        .timeStamp(startTime)
                        .build();


                CompletableFuture<Object> completableFuture = new CompletableFuture<>();
                // 需要将completableFuture暴露出去

                YrpcBootstrap.PENDING_REQUEST.put(yrpcRequest.getRequestId(), completableFuture);

                channel.writeAndFlush(yrpcRequest).addListener((ChannelFutureListener) promise -> {
                    if (!promise.isSuccess()) {
                        // 获取异常
                        completableFuture.completeExceptionally(promise.cause());
                    }
                });

                Long endTime = 0L;
                try {
                    // 阻塞方法，get()方法得不到结果就会一直阻塞
                    // 我们想不一直阻塞可以添加参数
                    completableFuture.get(1, TimeUnit.SECONDS);
                    endTime = System.currentTimeMillis();
                } catch (InterruptedException | ExecutionException | TimeoutException e) {

                    // 发生异常，可以进行重试3次，如果3次都失败，则将该主机从缓存中剔除
                    tryStamp--;
                    log.error("和地址{}的主机连接发生异常。正在进行第{}次重试", channel.remoteAddress(),3-tryStamp);
                    if (tryStamp == 0){

                        // 将断开连接的主机从缓存中剔除
                        YrpcBootstrap.CHANNEL_CACHE.remove(channel.remoteAddress());
                    }
                    // 尝试等到一段时间后再重试
                    try {
                        Thread.sleep(10*(new Random().nextInt(5)));
                    } catch (InterruptedException interruptedException) {
                        throw new RuntimeException(e);
                    }
                    continue;
                }
                Long time = endTime - startTime;
                // 使用treeMap进行缓存
                YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.put(time, channel);
                log.debug("响应的时间是{}", time);
                break;
            }
            }
            log.info("-----------------响应时间的TreeMap--------------------");
            for (Map.Entry<Long,Channel> entry : YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.entrySet()){
                if (log.isDebugEnabled()){
                    log.debug("响应时间是{}，对应的channel的id是{}",entry.getKey(),entry.getValue().id());
                }
            }

        }
    }
}
